package org.niit.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object SparkSQL_1 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sparkContext: SparkContext = spark.sparkContext
    sparkContext.setLogLevel("ERROR")

    //1.DataFrame读取Json文件
    val df: DataFrame = spark.read.json("input/user.json")
    df.show()
    //2.DataFrame使用SQL语法进行操作
    //创建一个张临时表 该表名叫做user
    df.createOrReplaceTempView("user")
    //2.1查询 user表下所有信息
    val resData1: DataFrame = spark.sql("select * from  user")
    //show 展示DataFrame内容
    resData1.show()
    //2.2查看DataFrame的结构信息
    df.printSchema()
    //2.3 查询 username 列 信息
    var resData2: DataFrame = spark.sql(" select username from user")
    resData2 = df.select("username")
    resData2.show()
    //2.4 查看 username列数据 以及 age + 1 的数据
    //DSL语法:管理结构化的数据一个特定领域的语言，使用DSL不用创建临时视图 。使用DSL语法需要引包
    import spark.implicits._
    df.select( $"username",$"age"+1 as "newAge" ).show()
    df.select( 'username,'age+1 as "newAge").show()

    //2.5查询出 age大于30的数据
    //df.select( $"age">30 ).show() //只会查询出结果是true还是false 。因为这里当作一个满足条件来去使用的
      df.filter($"age">30).show()

    //2.6 按照 age 进行分组，分组后显示每组数据的个数
    df.groupBy("age").count().show()

    //2.7 DataFrame 转换成 RDD 重点  将DataFrame中的字段名去掉 就变成了RDD
    val dataFrame: DataFrame = df.toDF()
    dataFrame.collect().foreach( println )
    val rdd1: RDD[Row] = df.rdd  // DataFrame === RDD[Row]
    rdd1.collect().foreach( println )


    //2.8 RDD 转换成 DataFrame 重点
    val rdd2: RDD[Int] = sparkContext.makeRDD((1 to 4).toList)
    val df2: DataFrame = rdd2.toDF("id")
    df2.show()

    //3.Dataset 是具有强类型的集合。需要提供对应的类型信息
    //3.1使用样例类创建Dataset       JDBC --> MySQL ---> 实体类
    val list = List(Person("张三", 30), Person("李四", 40))
    val ds1: Dataset[Person] = list.toDS()
    ds1.show()
    //3.2使用基本数据类型创建Dataset
    val list1: List[Int] = (1 to 10).toList
    val ds2: Dataset[Int] = list1.toDS()
    ds2.show()
    //3.3DataSet 转换成 RDD 重点
    val rdd3: RDD[Person] = ds1.rdd
    rdd3.collect().foreach( println )
    //rdd3.foreach( println )

    //3.4 RDD 转成 Dataset
    val ds3: Dataset[Person] = rdd3.toDS()
    ds3.show()

    //RDD 转成 Dataset 特殊方式
    val rdd4: RDD[(String, Int)] = sparkContext.makeRDD(List(  ("王五", 50), ("刘能", 60),("谢广坤", 70) )  )
    val ds4: Dataset[Person] = rdd4.map {
      case (name, age) => {
        Person(name, age)
      }
    }.toDS()
    ds4.show()

    //3.5DataFrame 和 Dataset的转换
    val df3: DataFrame = ds1.toDF()
    df3.show()
    val ds5: Dataset[Person] = df3.as[Person]
    ds5.show()

    //关闭环境
    spark.close()
  }



  //样例类
  case class Person(name:String,age:Int)
}
